package com.fastmms;

import com.fastmms.analysis.AnalysisBlock;
import com.fastmms.analysis.AnalysisBlockResult;
import com.fastmms.analysis.AnalysisFileUtil;
import com.fastmms.analysis.AnalysisMessage;
import com.fastmms.generate.GenerateMessageBody;
import com.fastmms.generate.GenerateUtil;

import java.nio.charset.StandardCharsets;
import java.util.*;

public class TestFindMsg {

    public static void main(String[] args) throws Exception{
        start();
    }

    public static void start() throws Exception{
        String msgfile = System.getProperty("user.dir") + "/testFile/1w_消息文件.mq";
        /**
         * 测试 生成消息数据
         */
        // 获取消息块信息并进行合并文件
        List<GenerateMessageBody> bodyList = create();
        try {
            GenerateUtil.merge(msgfile, bodyList);
        } catch (Exception e) {
            e.printStackTrace();
        }

        // 获取总的索引数据
        AnalysisBlockResult indexResult = AnalysisFileUtil.analysisIndex( AnalysisFileUtil.getIndexPath(msgfile) );
        /**
         * 测试代码块 解析消息数据
         */
        Scanner scanner = new Scanner(System.in);
        while (true){
            System.out.print("输入需要查找的消息开始序号：");
            String line = scanner.nextLine();
            System.out.print("输入限制多少行（查询该序号后的数据则为-1）：");
            String limit = scanner.nextLine();
            analysis(msgfile,indexResult,Integer.parseInt(line) , Integer.parseInt(limit));
        }
    }


    public static List<GenerateMessageBody> create(){
        int bodeSize = 10;
        List<GenerateMessageBody> bodyList = new ArrayList<>(bodeSize);
        int size = 0;
        long msgStartNo = 1;
        for (int i = 0; i < bodeSize; i++) {
            try {
                GenerateMessageBody body = new GenerateMessageBody();
                String subPath = System.getProperty("user.dir") + "/testFile/" + UUID.randomUUID()  +  ".temp";
                body.start(subPath , i,msgStartNo);
                for (int j = 0; j < 10000000; j++) {
                    size++;
                    String value = String.valueOf(System.currentTimeMillis() / ((Math.random() * 100000) + 1));
                    body.add(("这是第" + size + "条数据").getBytes(StandardCharsets.UTF_8), value.getBytes(StandardCharsets.UTF_8));
                }
                msgStartNo = body.end();
                bodyList.add(body);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        return bodyList;
    }

    public static void analysis(String msgfile , AnalysisBlockResult indexResult,long startNo , int limit ){
        long start = System.currentTimeMillis();
        AnalysisFileUtil.handleFile(msgfile, indexResult,startNo,limit,null, new AnalysisMessage() {
            @Override
            public void ready(List<AnalysisBlock> analysisBlockList, Object param) {
                System.out.println("开始执行消息块读取。消息块数量：" + analysisBlockList.size());
                for(AnalysisBlock block : analysisBlockList){
                    System.out.println("开始执行消息块读取。消息块开始地址 ：" + block.getStartPoint() +" / 消息块结束地址：" + block.getEndPoint() );
                }
            }

            @Override
            public void apply(long blockNo, long msgNo,long msgLingNo, byte[] keyData, byte[] valueData, Object param) {
                // 最好是输出到文档中，因为控制台会截断日志 导致无法判断是否正常解析
                String msg =  "消息块序号：" + blockNo + "/ 消息块序号：" + msgNo + "/ 消息行号：" + msgLingNo +  " - key : " + new String(keyData, StandardCharsets.UTF_8) + "  value : "
                        + new String(valueData,StandardCharsets.UTF_8);
                System.out.println(msg);
            }

            @Override
            public void finish(long readCount, Object param) {
                System.out.println("文件读取结束，已读取消息： " + readCount + "条，耗时：" + (System.currentTimeMillis() - start) + "毫秒");
                System.out.println();
            }

            @Override
            public void error(AnalysisBlock errorBlock, long errorPoint, Object param, Exception e) {
                System.out.println("读取消息块失败！读取错误点：" + errorPoint + (errorBlock != null ? ("错误消息块：" + errorBlock) : "")
                        + "，异常消息： " + e.getMessage());
                e.printStackTrace();
            }
        });
    }

}
